package io.openmessaging.demo;

import com.markhsiu.store.StoreDB;

import io.openmessaging.BatchToPartition;
import io.openmessaging.BytesMessage;
import io.openmessaging.KeyValue;
import io.openmessaging.Message;
import io.openmessaging.MessageFactory;
import io.openmessaging.Producer;
import io.openmessaging.Promise;

public class DefaultProducer  implements Producer {
	   
	    private MessageFactory messageFactory = new DefaultMessageFactory();
	    private KeyValue properties;

	    public DefaultProducer(KeyValue properties) {
	        this.properties = properties;
	        StoreDB.builder(properties, true);     
	    }


	    /**
		 * buidler
		 */
	    @Override 
	    public BytesMessage createBytesMessageToTopic(String topic, byte[] body) {
	        return messageFactory.createBytesMessageToTopic(topic, body);
	    }

	    /**
		 * buidler
		 */
	    @Override 
	    public 
	    BytesMessage createBytesMessageToQueue(String queue, byte[] body) {
	        return messageFactory.createBytesMessageToQueue(queue, body);
	    }

	    @Override 
	    public void start() {

	    }

	    @Override 
	    public void shutdown() {

	    }

	    @Override 
	    public KeyValue properties() {
	        return properties;
	    }

	    /**
		 * buidler
		 */
	    @Override 
	    public void send(Message message) {
			
	    	StoreDB.send((DefaultBytesMessage) message);
	    }

	    @Override public void send(Message message, KeyValue properties) {
	        throw new UnsupportedOperationException("Unsupported");
	    }

	    @Override 
	    public Promise<Void> sendAsync(Message message) {
	        throw new UnsupportedOperationException("Unsupported");
	    }

	    @Override 
	    public Promise<Void> sendAsync(Message message, KeyValue properties) {
	        throw new UnsupportedOperationException("Unsupported");
	    }

	    @Override 
	    public void sendOneway(Message message) {
	        throw new UnsupportedOperationException("Unsupported");
	    }

	    @Override 
	    public void sendOneway(Message message, KeyValue properties) {
	        throw new UnsupportedOperationException("Unsupported");
	    }

	    @Override 
	    public BatchToPartition createBatchToPartition(String partitionName) {
	        throw new UnsupportedOperationException("Unsupported");
	    }

	    @Override 
	    public BatchToPartition createBatchToPartition(String partitionName, KeyValue properties) {
	        throw new UnsupportedOperationException("Unsupported");
	    }


		@Override
		public void flush() {
			StoreDB.flush();
		}
}
